package edu.npu.GraphTransform;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;


public class GraphPartSig {
	public static final String SEPARATOR = "\t";
	public static final String FLAG = "#";
	
	//将key加上partition的signature
	public static class Key_GraphSigMapper extends 
			Mapper<LongWritable, Text, LongWritable, Text> {
		@Override
		protected void map(LongWritable key, Text value, Context context)
				throws IOException, InterruptedException {
			long k;
			String line = value.toString();
			String[] splits = line.split(SEPARATOR);
			if (splits[0].startsWith("#")) {
				k = Long.parseLong(splits[0].substring(1));
				context.write(new LongWritable(k), new Text("#" + splits[1]));
			} else {
				k = Long.parseLong(splits[0]);
				context.write(new LongWritable(k), new Text(splits[1]));
			}
		}
	}

	public static class Key_GraphSigReducer extends
			Reducer<LongWritable, Text, Text, Text> {
		@Override
		protected void reduce(LongWritable key, Iterable<Text> values,
				Context context) throws IOException, InterruptedException {
			Text partKey = null;
			List<Text> list = new ArrayList<Text>();
			for (Text value : values) {
				if (value.toString().startsWith("#")) {
					partKey = new Text(value.toString().substring(1));
					break;
				} else
					list.add(new Text(value));
			}
			for (Text value : values) 
				context.write(new Text(key.get() + "#" + partKey), value);
			for(Text value : list)
				context.write(new Text(key.get() + "#" + partKey), value);
		}
	}
	//将value加上partition的signature
	public static class Value_GraphSigMapper extends
			Mapper<LongWritable, Text, LongWritable, Text> {
		@Override
		protected void map(LongWritable key, Text value, Context context)
				throws IOException, InterruptedException {
			long v;
			String line = value.toString();
			String[] splits = line.split(SEPARATOR);
			if (splits[0].startsWith("#")) {
				v = Long.parseLong(splits[0].substring(1));
				context.write(new LongWritable(v), new Text("#" + splits[1]));
			} else {
				v = Long.parseLong(splits[1]);
				context.write(new LongWritable(v), new Text(splits[0]));
			}
		}
	}

	public static class Value_GrapSigReducer extends
			Reducer<LongWritable, Text, Text, Text> {
		@Override
		protected void reduce(LongWritable key, Iterable<Text> values,
				Context context) throws IOException, InterruptedException {
			Text partValue = null;
			List<Text> list = new ArrayList<Text>();
			for (Text value : values) {
				if (value.toString().startsWith("#")) {
					partValue = new Text(value.toString().substring(1));
					break;
				} else
					list.add(new Text(value));
			} 
			for (Text value : values) 
				context.write(value, new Text(key.get() + "#" + partValue));
			for(Text value : list)
				context.write(value, new Text(key.get() + "#" + partValue));
		}
	}
}
